package com.semidata.rtd.core.job;

import java.util.concurrent.ConcurrentLinkedQueue;

import com.semidata.rtd.core.service.SendableService;

public class JobTimeoutService implements SendableService<Job<?>> {
    
    public static final String NAME = "JOB_TIMEOUT";

    public static final long TIMEOUT = 100;
    private ConcurrentLinkedQueue<JobWrapper> queue = new ConcurrentLinkedQueue<>();
    private Thread thread = null;
    private boolean quit = false;
    private long timeout;

    private static class JobWrapper {
        private long timestamp;
        private Job<?> job;

        public JobWrapper(Job<?> job) {
            this.job = job;
            this.timestamp = System.currentTimeMillis();
        }

        public Job<?> getJob() {
            return job;
        }

        public long getTimestamp() {
            return timestamp;
        }
    }

    public JobTimeoutService() {
        this(TIMEOUT);
    }

    public JobTimeoutService(long timeout) {
        this.timeout = timeout;
    }

    @Override
    public void start() {
        thread = new Thread(new Runnable() {

            @Override
            public void run() {
                JobWrapper wrapper;
                while (!quit) {

                    try {
                        long now = System.currentTimeMillis();
                        long nextWake;

                        while (true) {
                            wrapper = queue.peek();

                            if (wrapper == null) {
                                nextWake = timeout;
                                break;
                            }

                            long timeElapse = now - wrapper.getTimestamp();
                            if (timeElapse < timeout) {
                                nextWake = timeout - timeElapse;
                                if (nextWake == 0)
                                    nextWake = 1;
                                break;
                            }
                            wrapper.getJob().cancel();
                            queue.remove();
                        }
                        Thread.sleep(nextWake);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                }

            }

        });
        thread.start();

    }

    @Override
    public void stop() {
        quit = true;
        try {
            if (thread != null)
                thread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void send(Job<?> job) {
        queue.add(new JobWrapper(job));
    }

}
